package com.example.listener;


import com.example.bean.Foo2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

/**
 * @author shawn
 * @since 2021/5/12 17:42
 */
@Component
public class KafkaListenerGroup {

    private final Logger LOGGER = LoggerFactory.getLogger(KafkaListenerGroup.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @KafkaListener(id = "fooGroup2", topics = "topic2")
    public void listen1(List<Foo2> foos) throws IOException {
        LOGGER.info("Received: " + foos);
        foos.forEach(f -> kafkaTemplate.send("topic3", f.getFoo().toUpperCase()));
        LOGGER.info(" commit tx");
        //由于上一次异常未处理会导致消费回滚，然后会一直启动一直报异常
//        throw new RuntimeException("运行时异常");
    }

    @KafkaListener(id = "fooGroup3", topics = "topic3")
    public void listen2(List<String> in) {
        LOGGER.info("Received: " + in);
    }
}
